package com.plugins.server;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DefaultConsumer;



public class MQServer {
	
	private static final int HEAR_BEAT = 5;
	private static MQServer sInstance;
	private ExecutorService threadPool;
	private ConnectionFactory factory;
	
	private MQServer(){
		this.factory = new ConnectionFactory();
		this.factory.setHost("127.0.0.1");
		this.factory.setUsername("yanghui");
		this.factory.setPassword("yanghui");
		this.factory.setRequestedHeartbeat(HEAR_BEAT);
		this.factory.setAutomaticRecoveryEnabled(true);
		this.threadPool = Executors.newFixedThreadPool(1);
		this.factory.setSharedExecutor(threadPool);
		this.factory.setExceptionHandler(new MessageQueueExceptionHandler());
	}
	
	public synchronized static MQServer getServer(){
		if(sInstance == null)
			sInstance = new MQServer();
		return sInstance;
	}
	
	
	public Channel initConnection(String exchangeName, String queueName, String routeKey){
		try{
			Connection connection = this.factory.newConnection();
			Channel channel = connection.createChannel();
			channel.exchangeDeclare(exchangeName, "direct", true);
			channel.queueDeclare(queueName, true, false, false, null);
			channel.queueBind(queueName, exchangeName, routeKey);
			int numCPUs = Runtime.getRuntime().availableProcessors();
			channel.basicQos(numCPUs*2, true);
			return channel;
		}catch(Exception e){
			e.printStackTrace();
			return null;
		}
		
	}
	
	public void startConsume(DefaultConsumer consumer, String queueName) {
		try{
			consumer.getChannel().basicConsume(queueName, false, consumer);
		}catch(Exception e){
			e.printStackTrace();
		}
	}
	
	public void ackMessage(Channel channel, Long deliveryTag) {
		try{
			channel.basicAck(deliveryTag, false);
		}catch(Exception e){
			e.printStackTrace();
		}
	}
	
	/**
	 * 发送消息
	 * @param queueName
	 * @param exchangeName
	 * @param routeKey
	 * @param msg
	 */
	public void publishMsg(String queueName, String exchangeName, String routeKey, String msg){
		try{
			Connection connection = this.factory.newConnection();
			Channel channel = connection.createChannel();
			channel.exchangeDeclare(exchangeName, "direct", true);
			channel.queueDeclare(queueName, true, false, false, null);
			channel.queueBind(queueName, exchangeName, routeKey);
			channel.basicPublish(exchangeName, routeKey, null, msg.getBytes());
			channel.close();
			connection.close();
		}catch(Exception e){
			e.printStackTrace();
		}
	}
}
